package org.niit.kafka;

import org.apache.kafka.clients.producer.*;
import org.apache.kafka.common.serialization.StringSerializer;

import java.util.Properties;

/*
  异步的方式去生产数据
     异步的方式会带有回调函数， 该函数包含两个参数，一个metaData，一个exception
                     作用：Kafka服务器获得了生产者的数据，并给生产者一个回馈，可能是成功 也可以是失败
 */
public class KafkaProducerTest2 {


    public static void main(String[] args) throws InterruptedException {
        //1.配置连接信息
        Properties props = new Properties();
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,"node1:9092");
        props.put(ProducerConfig.ACKS_CONFIG,"all");
        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,StringSerializer.class.getName());


        //2.创建生产者对象
        KafkaProducer<String,String> producer = new KafkaProducer<String, String>(props);

        //3.生产数据
        for (int i = 0;i < 100000;i++){
            //3.1实例化一个数据对象
            ProducerRecord<String,String> record = new ProducerRecord<>("BD2",null,i+"模拟");

            producer.send(record, new Callback() {
                @Override
                public void onCompletion(RecordMetadata recordMetadata, Exception e) {
                    //3.2判断是否发送成功
                    if(e ==null ){
                        //发送成功
                        String topic = recordMetadata.topic();
                        int partition = recordMetadata.partition();
                        long offset = recordMetadata.offset();
                        long timestamp = recordMetadata.timestamp();

                        System.out.println("发送成功：主题："+topic+",分区："+partition+",偏移量："+offset+",时间戳："+timestamp);

                    }else{
                        System.out.println("发送失败："+e.getMessage());
                    }
                }
            });
            Thread.sleep(1000);
        }
        //4.关闭生产者
        producer.close();
    }

}
